package com.project.pulsar.persistent;

import com.project.pulsar.conf.PulsarConf;
import org.apache.pulsar.client.admin.Namespaces;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 持久化操作
 */
@RestController
public class PersistentController {
    @Autowired
    PulsarConf pulsarConf;

    /**
     * 生产消息
     * @param msg
     * @param tag 0默认持久化的消息 1非持久化消息 2 持久化消息的另外一个写法，同0
     * @throws PulsarClientException
     */
    @GetMapping("/persistent/sendMsg")
    public MessageId sendMsg(String msg, String tag) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic="my-topic";
        if("1".equals(tag)) {
            topic="non-persistent://public/default/my-topic";
                   //非持久化         租户   命名空间   主题
        }
        if("2".equals(tag)){
            topic="persistent://public/default/my-topic";
                  //持久化         租户   命名空间   主题
        }
        Producer<byte[]> producer1 = pulsarFactory.newProducer()
                .topic(topic)
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        return producer1.send(msg.getBytes());
    }

    /**
     * 手动执行获取消息
     * @param tag 0默认持久化的消息 1非持久化消息 2 持久化消息的另外一个写法，同0
     * @throws PulsarClientException
     */
    @GetMapping("/persistent/comsumer")
    public void comsumerByArtificial(String tag) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic="my-topic";
        if("1".equals(tag)) {
            topic="non-persistent://public/default/my-topic";
                   //非持久化         租户   命名空间   主题
        }
        if("2".equals(tag)){
            topic="persistent://public/default/my-topic";
                 //持久化         租户   命名空间   主题
        }
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }


    /**
     * 创建租户
     * @throws Exception
     */
    @GetMapping("/persistent/createTenant")
    public void createTenant(String tenantName,String clusterName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            Set set = new HashSet();
            set.add(clusterName);
            TenantInfo build = TenantInfo.builder().allowedClusters(set).build();
            tenants.createTenant(tenantName, build);
        }
    }
    /**
     * 获得现有租户
     * @throws Exception
     */
    @GetMapping("/persistent/tenantList")
    public List<String> tenantList() throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            return tenants.getTenants();
        }
        return null;
    }
    /**
     * 修改现有租户配置(所属集群等信息)
     * @throws Exception
     */
    @GetMapping("/persistent/updateTenant")
    public void updateTenant(String tenantName,String clusterName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            Set set = new HashSet();
            set.add(clusterName);
            TenantInfo build = TenantInfo.builder().allowedClusters(set).build();
            tenants.updateTenant(tenantName, build);
        }
    }
    /**
     * 删除指定租户
     * @throws Exception
     */
    @GetMapping("/persistent/delTenant")
    public void delTenant(String tenantName) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Tenants tenants = admin.tenants();
            tenants.deleteTenant(tenantName);
        }
    }

    /**
     * 创建命名空间
     * @param namespace 租户名字/命名空间
     * @throws Exception
     */
    @GetMapping("/persistent/createNamespace")
    public void createNamespace(String namespace) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            namespaces.createNamespace(namespace);
        }
    }

    /**
     * 根据租户获得命名空间
     * @param tenant
     * @return
     * @throws Exception
     */
    @GetMapping("/persistent/namespaceList")
    public List<String> namespaceList(String tenant) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            return namespaces.getNamespaces(tenant);
        }
        return null;
    }

    /**
     * 删除命名空间
     * @param npname
     * @throws Exception
     */
    @GetMapping("/persistent/delNamespace")
    public void delNamespace(String npname) throws Exception {
        PulsarAdmin admin = pulsarConf.pulsarAdminFactory();
        if(admin!=null) {
            Namespaces namespaces = admin.namespaces();
            namespaces.deleteNamespace(npname);
        }
    }

}
